package com.lry.raft;

import com.lry.BrokerController;
import com.lry.DefaultMessageStore;
import com.lry.core.StateMachine;
import com.lry.message.AckMessage;
import com.lry.message.Message;
import com.lry.message.MessageExt;
import com.lry.message.PullMessage;
import com.lry.producer.SendResult;
import com.lry.serialize.KryoSerializer;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.CompletableFuture;

@Slf4j
public class MqStateMachine implements StateMachine {

    private BrokerController brokerController;

    public MqStateMachine(BrokerController brokerController) {
        this.brokerController = brokerController;
    }

    @Override
    public void writeSnapshot(String snapshotDir) {

    }

    @Override
    public void readSnapshot(String snapshotDir) {

    }

    @Override
    public byte[] apply(byte[] dataBytes) {
        Message msg = KryoSerializer.deserialize(dataBytes);
        SendResult sendResult = brokerController.produce(msg);
        return KryoSerializer.serialize(sendResult);
    }

    @Override
    public byte[] get(byte[] command) {
        PullMessage msg = KryoSerializer.deserialize(command);
        List<MessageExt> messageExts = brokerController.consume(msg,true);
        return KryoSerializer.serialize(messageExts);
    }

    @Override
    public void tobeMaster(boolean isMaster) {
        CompletableFuture.runAsync(()->{
            brokerController.tobeMaster(isMaster);
        });
    }


}
